package com.cqwshzj.backgroundjob.job;/**
 * @Description:
 * @Auther: Huang Zhijin
 * @Version: 1.0.0
 * @Date: 2021/4/7 18:43
 */

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.cqwshzj.aware.AppContextHolder;
import com.cqwshzj.common.JobConstant;
import com.cqwshzj.common.column.JobInfoColumn;
import com.cqwshzj.framework.dao.api.JobTypeDao;
import com.cqwshzj.framework.dao.model.JobInfo;
import com.cqwshzj.framework.dao.model.JobStrategy;
import com.cqwshzj.framework.dao.model.JobType;
import com.cqwshzj.framework.exception.unchecked.JobFrameworkException;
import com.cqwshzj.framework.job.AbstractSimpleJob;
import com.cqwshzj.framework.service.JobTypeService;
import com.cqwshzj.framework.service.model.AddJobParam;
import com.cqwshzj.util.CrontabUtil;
import com.cqwshzj.util.SysUtils;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 * 扫描任务策略动态产生任务信息
 * @author Huang Zhijin
 * @create 2021/4/7
 * @since 1.0.0
 */
public class ScanJobStrategyAndDynamicAddJob extends AbstractSimpleJob implements SimpleJob {

	private static final Logger log = LoggerFactory.getLogger( ScanJobStrategyAndDynamicAddJob.class );

	private int jobAddCount = 0;

	//-----------------

	private JobTypeDao jobTypeDao;
	private JobTypeService jobTypeService;

	//-----------------

	private Map<Integer, Consumer<JobStrategy>> jobTypeStrategyConsumerMap = new HashMap<>();

	private Consumer<JobStrategy> jobStrategyExceptionHandlerConsumer = (obj)->{
		try {
			jobTypeStrategyConsumerMap.get( obj.getStrtType() ).accept( obj );
		}catch (Exception e){
			log.error(" add jobInfo by jobStrategy ERROR. jobStrategy:[ "+ SysUtils.obj2JsonStr(obj) +" ] . exception: "+e.getMessage(),e);
			jobStrategyService.updateStatusEnableById( obj.getId() );
		}
	};

	//-----------------

	@Override
	public void execute(ShardingContext shardingContext) {
		super.executeJob(shardingContext);
	}

	@Override
	public void initInstanceProperties(ShardingContext shardingContext) {
		super.initInstanceProperties(shardingContext);

		//---------
		jobTypeService = AppContextHolder.getBean( JobTypeService.class);
		jobTypeDao = AppContextHolder.getBean( JobTypeDao.class);

		//---------------
		jobTypeStrategyConsumerMap.put( null, this::noOperationJobInfoAdder );
		jobTypeStrategyConsumerMap.put(JobConstant.JOB_STRATEGY_TYPE_ONCE_A_DAY, this::onceADayJobInfoAdder );
		jobTypeStrategyConsumerMap.put(JobConstant.JOB_STRATEGY_TYPE_INTERVAL, this::todayIntervalJobInfoAdder );
		jobTypeStrategyConsumerMap.put(JobConstant.JOB_STRATEGY_TYPE_ONLY_ONCE, this::onlyOnceJobInfoAdder );

	}

	@Override
	public void doExecute(ShardingContext shardingContext) {
		jobAddCount = 0;
		long start = System.currentTimeMillis();
		log.info(" scan jobstrategy start.  startTime=[{}] ",start );

		final List<JobStrategy> list = jobStrategyService.listEnableStrategyOfToday();
		if (CollectionUtils.isEmpty(list)) {
			log.debug(" job_strategy has no enabled record in today. today=[{}] ", DateFormatUtils.format(new Date(), JobConstant.DATE_2) );
			return;
		}
 		list.forEach( jobStrategyExceptionHandlerConsumer::accept );
		log.info(" scan jobstrategy successfully. added {} jobs . total cost [{}] ms.",jobAddCount,(System.currentTimeMillis()-start) );
	}


	//-------------------------------------

	/**
	 *
	 * @param obj
	 */
	public void noOperationJobInfoAdder(JobStrategy obj){
	}

	/**
	 * 一日只执行一次的任务创建
	 * @param obj
	 */
	public void onceADayJobInfoAdder( JobStrategy obj){
		Assert.notNull( obj, "缺少 jobStrategy");
		Assert.isTrue( obj.getStrtType() != null && JobConstant.JOB_STRATEGY_TYPE_ONCE_A_DAY == obj.getStrtType().intValue() ,"缺少参数 strtType 或 strtType 不等于 1");
		//--1.校验参数、校验任务是否已存在
		//--2.准备参数并添加任务
		if ( !checkNormalJobExistAndDynamicAddNormalJob(obj)) {
			return;
		}

		//--4.修改 NEXT_RUN_DATE +1
		DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern( JobConstant.DATE_1 );
		LocalDate currentRunDate = LocalDate.parse( obj.getNextRunDate() , dateFormatter );
		LocalDate nextRunDate = currentRunDate.plusDays( 1L );

		jobStrategyService.updateNextRunDateAndStatuaEnableById(obj.getId(), nextRunDate.format(dateFormatter) );
		jobAddCount++;
	}

	/**
	 * 一日只执行一次的任务创建
	 * @param obj
	 */
	public void onlyOnceJobInfoAdder( JobStrategy obj){
		Assert.notNull( obj, "缺少 jobStrategy");
		Assert.isTrue( obj.getStrtType() != null && JobConstant.JOB_STRATEGY_TYPE_ONLY_ONCE == obj.getStrtType().intValue() ,"缺少参数 strtType 或 strtType 不等于 4");

		//--1.校验参数、校验任务是否已存在
		//--2.准备参数并添加任务
		if ( !checkNormalJobExistAndDynamicAddNormalJob(obj)) {
			return;
		}

		//--4.修改 NEXT_RUN_DATE +1
		DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern( JobConstant.DATE_1 );
		LocalDate currentRunDate = LocalDate.parse( obj.getNextRunDate() , dateFormatter );
		LocalDate nextRunDate = currentRunDate.plusYears( 1000L );

		jobStrategyService.updateNextRunDateAndStatuaEnableById(obj.getId(), nextRunDate.format(dateFormatter) );
		jobAddCount++;
	}


	/**
	 * 一日只执行一次的任务创建
	 * @param obj
	 */
	public void todayIntervalJobInfoAdder( JobStrategy obj){
		Assert.notNull( obj, "缺少 jobStrategy");
		Assert.isTrue( obj.getStrtType() != null && JobConstant.JOB_STRATEGY_TYPE_INTERVAL == obj.getStrtType().intValue() ,"缺少参数 strtType 或 strtType 不等于 3");
		Assert.hasText( obj.getNextRunDate(), "缺少参数 nextRunDate");
		Assert.hasText( obj.getStartTime(), "缺少参数 startTime");
		Assert.hasText( obj.getIntvMin(), "缺少参数 intvMin");
		Assert.notNull( obj.getJobType(), "缺少参数 jobType");
		Assert.notNull( obj.getStrtType(), "缺少参数 strtType");
		Assert.notNull( obj.getTimeWindow(), "缺少参数 timeWindow");

		//--0.对当前的 jobStrategy 加上行锁
		try {
			jobStrategyService.lockAndUpdateStateDisabledById(obj.getId(), obj.getNextRunDate() );
		}catch (Exception e){
			log.info(" jobStrategy record locked . skip add job . jobStrategy: {} .",SysUtils.obj2JsonStr(obj) );
			return;
		}

		DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern( JobConstant.DATE_1 );
		DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern( JobConstant.TIME_4 );

		//--2.计算任务执行的日期、执行周期、cron表达式
		LocalDate currentRunDate = LocalDate.parse(obj.getNextRunDate(), dateFormatter );
		LocalTime startTime = LocalTime.parse( obj.getStartTime(), timeFormatter );
		LocalTime endTime = startTime.plusMinutes( obj.getTimeWindow() );
		//周期性任务 分钟数
		int intervalMin = Integer.parseInt( obj.getIntvMin().trim() );

		String intervalCron = CrontabUtil.getIntervalMinuteCrontabOfTargetDate(currentRunDate, intervalMin, startTime.getHour(), endTime.getHour() );
		//--3.准备参数并添加任务
		prepareNormalParameterAndDynamicAndJob(obj, obj.getNextRunDate(), intervalCron );

		//--4.更新 job_strategy ，修改 next_run_Date +1
		LocalDate nextRunDate = currentRunDate.plusDays(1L);

		jobStrategyService.updateNextRunDateAndStatuaEnableById(obj.getId(), nextRunDate.format(dateFormatter) );
		jobAddCount++;
	}


	/**
	 * 检查任务是否存在，并添加任务到zk中
	 * @param obj
	 * @return
	 */
	private boolean checkNormalJobExistAndDynamicAddNormalJob(JobStrategy obj){
		Assert.notNull( obj, "缺少参数");
		Assert.notNull( obj.getJobType(), "缺少参数 jobType");
		Assert.notNull( obj.getStrtType() , "缺少参数 strtType");
		Assert.hasText( obj.getNextRunDate(), "缺少参数 nextRunDate");
		Assert.hasText( obj.getStartTime(), "缺少参数 startTime");

		//--0.对当前 jobstrategy 加锁
		try {
			jobStrategyService.lockAndUpdateStateDisabledById( obj.getId(), obj.getNextRunDate() );
		}catch (Exception e){
			log.info(" jobStrategy is already locked.  {}", SysUtils.obj2JsonStr(obj) );
//			log.error(e.getMessage(),e);
			return false;
		}

		//--1.检查任务是否存在
		/*if ( checkNormalJobExists(obj) ){
			jobStrategyService.updateStatusEnableById( obj.getId() );
			return false;
		}*/

		DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern( JobConstant.DATE_1 );
		DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern( JobConstant.TIME_4 );
		DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern( JobConstant.JOB_DATETIME );

		//--2.检查当前时间 是否存在于策略的任务执行窗口期中。
		LocalDate nextRunDate = LocalDate.parse( obj.getNextRunDate() , dateFormatter );
		LocalTime jobStartTime = LocalTime.parse( obj.getStartTime(), timeFormatter );
		LocalDateTime jobStartDateTime = LocalDateTime.of( nextRunDate, jobStartTime );
		LocalDateTime jobEndDateTime = jobStartDateTime.plusMinutes( obj.getTimeWindow() == null ? JobConstant.DEFAULT_TIME_WINDOW : obj.getTimeWindow() );
		LocalDateTime now = LocalDateTime.now();

		if (now.isAfter( jobEndDateTime )) {
			log.info( " job's time windows passed. create job instance error. now=[{}] jobEndtIME=[{}] . {}",now.format(dateTimeFormatter), jobEndDateTime.format(dateTimeFormatter), SysUtils.obj2JsonStr(obj));
			return false;
		}

		//--3.准备参数并添加任务
		//如果当前时间已在 任务的执行窗口期中了，那么任务执行时间为 当前时间 + 随机延迟
		//否则为 任务执行时间为 job_strategy 配置的时间
		if (now.isBefore(jobStartDateTime)) {
			LocalDateTime exeTime = LocalDateTime.of( nextRunDate, jobStartTime ).plusSeconds( RandomUtils.nextInt( JobConstant.DELAY_SECONDS_BEGIN, JobConstant.DELAY_SECONDS_END) );
			String cron = CrontabUtil.getCrontabByLocalDateTime( exeTime );
			prepareNormalParameterAndDynamicAndJob(obj, obj.getNextRunDate(), cron );
		}else{
			LocalDateTime exeTime = now.plusSeconds( RandomUtils.nextInt( JobConstant.DELAY_SECONDS_BEGIN, JobConstant.DELAY_SECONDS_END) );
			String cron = CrontabUtil.getCrontabByLocalDateTime( exeTime );
			prepareNormalParameterAndDynamicAndJob(obj, obj.getNextRunDate(), cron );
		}
		return true;
	}


	/**
	 * 准备任务的参数，并动态添加到 zk 中
	 * @param obj
	 * @param occurDate
	 * @param cron
	 */
	private void prepareNormalParameterAndDynamicAndJob(JobStrategy obj,
														String occurDate,
														String cron) {
		Assert.notNull( obj,"缺少参数 jobStrategy ");
		Assert.hasText( occurDate,"缺少参数 occurDate ");
		Assert.hasText( cron,"缺少参数 cron ");
		//--1.准备参数
		final JobType jobType = jobTypeService.selectOneByJobType(obj.getJobType());
		Map<String,Object> strtParamMap = SysUtils.jsonStr2Obj( obj.getParams(), HashMap.class );
		// jobStrategy 没有任务参数
		if ( CollectionUtils.isEmpty(strtParamMap) ) {
			strtParamMap = new HashMap<>(4);
		}

		String jobId = obj.getJobType() +"_"+ DateFormatUtils.format(new Date(), JobConstant.DATETIME_2)+ RandomUtils.nextInt(10,100);
		int shardingTotalCount = MapUtils.getIntValue( strtParamMap, JobConstant.PARAMS_KEY_SHARDING_TOTAL_COUNT, 1 );

		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < shardingTotalCount; i++) {
			if (i>0){
				sb.append(",");
			}
			sb.append(i).append("=").append(jobId);
		}

		Class<? extends SimpleJob> simpleJobClass = null;
		try{
			simpleJobClass = (Class<? extends SimpleJob>) ClassUtils.forName( jobType.getClassName(), ScanJobStrategyAndDynamicAddJob.class.getClassLoader() );
		}catch (Exception e){
			log.error(" scan jobStrategy error, failed to load SimpleJob Class. "+e.getMessage(),e);
		}

		String jobName = jobId;

		AddJobParam param = new AddJobParam();
		param.setJobId( jobId );
		param.setParentJobId( null );
		param.setJobType( obj.getJobType() );
		param.setStrategyId( obj.getId() );
		param.setParams( strtParamMap );
		//elastic-job 参数
		param.setJobName( jobName );
		param.setSimpleJobClass( simpleJobClass );
		param.setCron( cron );
		param.setShardingTotalCount((short) shardingTotalCount);
		param.setShardingItemParameters( sb.toString() );

		try {
			jobInfoService.dynamicAddJob( param );
		}catch (IllegalArgumentException | JobFrameworkException e){
			log.error(" add job into elastic-job error. jobStrategy: "+SysUtils.obj2JsonStr(obj)+" . message: "+e.getMessage(),e);
			throw e;
		}
	}


	private boolean checkNormalJobExists(JobStrategy obj){
		QueryWrapper<JobInfo> wrapper = new QueryWrapper<>();
		wrapper.eq( JobInfoColumn.STRT_ID, obj.getId() );
		wrapper.eq( JobInfoColumn.JOB_TYPE, obj.getJobType() );
		wrapper.likeRight( JobInfoColumn.JOB_UPD_TIME, obj.getNextRunDate() );

		final List<JobInfo> jobInfos = jobInfoDao.selectList(wrapper);
		if (CollectionUtils.isEmpty( jobInfos )) {
			return true;
		}
		return false;
	}



}
